package com.fitness.front.service.impl.common;

import cn.hutool.json.JSONUtil;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executor;

@Slf4j
@Service
@RequiredArgsConstructor
public class FlumeExecutorServiceImpl {

    @Value("${flume.ip}")
    private String ip;
    @Value("${flume.port}")
    private int port;

    private final Executor sendRateExecutor;

    public void sendRate(Long uid,Long cid,double rate) {
        sendRateExecutor.execute(new SendRateTask(uid,cid,rate));
    }

    @AllArgsConstructor
    class SendRateTask implements Runnable {
        private Long uid;
        private Long cid;
        private double rate;

        @Override
        public void run() {
            RpcClient client = RpcClientFactory.getDefaultInstance(ip, port);
            try {
                client.append(EventBuilder.withBody(uid+","+cid+rate+System.currentTimeMillis(), StandardCharsets.UTF_8));
                log.info("用户评分：{}", uid+","+cid+","+rate+","+System.currentTimeMillis());
            } catch (EventDeliveryException e) {
                throw new RuntimeException(e);
            }
            client.close();
        }
    }
}
